/**
 * @file mqttmgr.cpp
 * @brief mqtt客户端封装，以单例的形式对外提供订阅发布功能。
 *
 *用mosquitto的订阅发布进行封装，线程安全
 *
 * @author ypj
 * @date 2024年3月21日
 * @license MIT Apache2.0
 * Copyright (C), 2010-2023 Jouav Co., Ltd.
 */
#include "mqttmgr.h"
#include "log.h"
#include "config.h"

namespace jouav
{
    MqttMgr::MqttMgr()
    {
        mosqpp::lib_init();
    }

    MqttMgr::~MqttMgr()
    {
        mosqpp::lib_cleanup();
    }
    /**
    * @brief 单例接口获取MQMgr的实例化对象
    * @return MQMgr实例化对象
    */
    MqttMgr& MqttMgr::GetInstance()
    {
        static MqttMgr mqttmgr;
        return mqttmgr;
    }
    /**
    * @brief 打印mqtt 配置日志
    * @return 
    */
    static void PrintMqttInfo()
    {
        log_sys->info("mqtt ip:{},port:{},name:{},id:{},qos:{}",
            jouav::Config::GetInstance().GetMqttIp(),
            jouav::Config::GetInstance().GetMqttPort(),
            jouav::Config::GetInstance().GetMqttName(),
            jouav::Config::GetInstance().GetMqttId(),
            jouav::Config::GetInstance().GetMqttQos());
        log_sys->info("mqtt subtopic:{}", jouav::Config::GetInstance().GetMqttSubTopic());
        log_sys->info("mqtt pubtopic:{}", jouav::Config::GetInstance().GetMqttPubTopic());
        return;
    }
    /**
        * @brief MQTT初始化
        * @return 0:成功 -1:失败
        */
    int MqttMgr::Init()
    {
        PrintMqttInfo();
        int res = 0;
        res = reinitialise(jouav::Config::GetInstance().GetMqttId().c_str(), true);
        if (res < 0)
        {
            log_sys->error("reinitialise failed. res {}", res);
            return -1;
        }

        if (jouav::Config::GetInstance().GetMqttName().size() > 0)
        {
            res = username_pw_set(jouav::Config::GetInstance().GetMqttName().c_str(),
                jouav::Config::GetInstance().GetMqttKey().c_str());
            if (res < 0)
            {
                log_sys->error("username_pw_set failed. res {}", res);
                return -1;
            }
        }
        //res = connect_async("114.116.226.82", 1883, 90);
        printf("mqtt ip:%s, port:%d\n", jouav::Config::GetInstance().GetMqttIp().c_str(), jouav::Config::GetInstance().GetMqttPort());
        res = connect_async(jouav::Config::GetInstance().GetMqttIp().c_str(), 
                jouav::Config::GetInstance().GetMqttPort());
        if (res < 0)
        {
            log_sys->error("connect_async failed. res {}", res);
            return -1;
        }

        res = loop_start();
        if (res)
        {
            log_sys->error("loop_start failed. res {}", res);
        }
        return res;
    }

    /*!
        * @brief 添加订阅
        * @param topic_ 订阅主题
        * @param func_ 回调函数
        * @return int: 0:没有错误
        * @note 回调函数返回
        */
    int MqttMgr::Subscribe(const std::string& topic_/*, mqttsub_cb func_*/)
    {

        static int mid = 0;
        mid++;
        int res = subscribe(&mid, topic_.c_str());
        //if (res == 0)
        //{
        //    map_callback_[topic_] = func_;
        //}
        return 0;
    }

    int MqttMgr::SetSubCb(mqttsub_cb func_)
    {
        cb_ = func_;
        return 0;
    }

    /*!
        * @brief 取消订阅
        * @param topic_ 取消订阅的Topic
        * @return int: 0:没有错误 -1:
        */

    int MqttMgr::UnSubscribe(const std::string& topic_)
    {
        static int mid = 0;
        mid++;
        return  unsubscribe(&mid, topic_.c_str());
    }

    /*!
    * @brief 发布消息
    * @param topic_ 发布主题
    * @param data_ 数据
    * @param len_ 长度 ，-1失败
    * @return int: 0:没有错误 -1:内部错误 65535：没有授权
    */
    int MqttMgr::Publish(const std::string& topic_, const void* data_, size_t len_)
    {
        if (is_connect_)
        {
            static int mid = 0;
            mid++;
            return publish(&mid, topic_.c_str(), len_, data_, jouav::Config::GetInstance().GetMqttQos());
        }
        return -1;
    }

    void MqttMgr::on_connect(int rc)
    {
        is_connect_ = true;
        log_sys->info("on_connect. rc {}", rc);
        Subscribe(jouav::Config::GetInstance().GetMqttSubTopic());
        return;
    }

    void  MqttMgr::on_connect_with_flags(int rc, int flags)
    {
        log_sys->info("on_connect_with_flags . rc {}, flags {}", rc, flags);
        return;
    }

    void  MqttMgr::on_disconnect(int rc)
    {
        is_connect_ = false;
        reconnect_async();
        log_sys->error("on_disconnect . res {}", rc);
        return;
    }

    void  MqttMgr::on_publish(int mid)
    {
        static uint64_t push_cnt = 0;
        if ((push_cnt % 100) == 0)
        {
            log_sys->info("on_publish . mid {}, success cnt:{}", mid, push_cnt);
        }
        push_cnt++;
        return;
    }

    void  MqttMgr::on_message(const struct mosquitto_message* message)
    {
        if (message != NULL)
        {
            /*auto itr = map_callback_.find(std::string(message->topic));
            if (itr != map_callback_.end())
            {
                auto cb = itr->second;
                cb(message->topic, message->payload, message->payloadlen);
            }*/
            if (cb_)
            {
                (*cb_)(message->topic, message->payload, message->payloadlen);
            }
        }
        return;
    }

    void  MqttMgr::on_subscribe(int mid, int qos_count, const int* granted_qos)
    {
        log_sys->info("on_subscribe . mid {}, qos_count {}", mid, qos_count);
        return;
    }

    void  MqttMgr::on_unsubscribe(int mid)
    {
        log_sys->info("on_unsubscribe . mid {}", mid);
        return;
    }

    void  MqttMgr::on_log(int level, const char* str)
    {
        log_sys->info("on_publish . level {}, {}", level, str);
        return;
    }

    void  MqttMgr::on_error()
    {
        return;
    }

}// namespace jouav
